package com.yinanxi.cool.dws

import com.yinanxi.cool.SparkUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD

import scala.collection.mutable.ListBuffer

/**
 * @Created by Orion
 * @Description
 *       目标事件 : 下单
 *         待归因事件 : productShow , addcar , 分享
 * 1 确认  目标事件      待归因事件
 * 2 留下目标事件的数据   关联    待归因事件
 * 3 对每个人分析 分组      对待归因事件 时间排序
 * 4 确认归因策略
 * 抽取明细表中的部分字段 , 提升计算效率 减少计算数据量
 *
 * 目标事件  :   e4  p2=v2
 * 待归因事件(没有指定顺序) :  [e1 p1=v1 , e2 , e3]
 */
object Attribution {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)

    val session = SparkUtils.getSesion()


    val df1 = session.sql(
      """
        |select
        |x2.*
        |from
        |     (select
        |guid
        |from  dws.cool_app_funnel_dtl
        |where dt = ' 2023-02-29'
        |and (event_id = 'e4'  and properties['p2'] = 'v2')
        |group by  guid) x1
        |join
        |         -- 2 待归因事件数据
        |(select *
        |from   dws.cool_app_funnel_dtl
        |where dt = ' 2023-02-29'
        |and (
        |    (event_id = 'e4'  and properties['p2'] = 'v2')
        |    or
        |    (event_id = 'e1'  and properties['p1'] = 'v1')
        |    or event_id = 'e3'
        |    or event_id = 'e2'
        |    )) x2
        |on x1.guid = x2.guid
        |""".stripMargin)

    val rdd1 = df1.rdd.map(row => {
      val guid = row.getAs[Long]("guid")
      val eventId = row.getAs[String]("event_id")
      val ts = row.getAs[Long]("ts")
      (guid, eventId, ts)
    })
// 分组
   val grouped = rdd1.groupBy(_._1)
    // 对分组内的事件数据进行排序
    val rdd2: RDD[(Long, ListBuffer[ListBuffer[String]])] = grouped.map(tp => {
      val guid = tp._1
      // 排序
      val list = tp._2.toList.sortBy(_._3)
      // 存储 目标事件可能出现多次
      val lists: ListBuffer[ListBuffer[String]] = ListBuffer[ListBuffer[String]]()
      // 每次,目标事件之间的待归因事件列表
      var events: ListBuffer[String] = ListBuffer[String]()
      //((1,e3,10), (1,e2,12), (1,e3,13), (1,e4,14), (1,e2,15), (1,e1,16), (1,e3,17), (1,e4,18), (1,e1,19)))
      list.map(data => {
        if (!data._2.equals("e4")) {
          events.append(data._2)
        } else {
          lists.append(events)
          events = ListBuffer[String]()
        }
      })
      (guid, lists.filter(_.nonEmpty))
    })

  // 对每个人的多个归因列表分析
  //  (1, List( List(e3,e2 ,e3) , List(e2,e1,e3)))
    println("================首次触点归因=================")
    rdd2.flatMap(tp=>{
      val guid = tp._1
      tp._2.map(list=>{
        // 事件去重  待归因事件只能做为一个位置的计算标准
        val list2 = list.distinct
        for (i <- 0 until list2.size) yield {
          if (i==0){
            (guid , list2(i) , 100)  // 首次触点
          }else {
            (guid , list2(i) , 0)
          }
        }
      })
    }).foreach(println)

    /**
     * 事件不去重分析
     * Vector((1,e3,100), (1,e2,0), (1,e3,0))
     * Vector((1,e2,100), (1,e1,0), (1,e3,0))
     * Vector((3,e2,100))
     * Vector((3,e3,100))
     * Vector((2,e3,100), (2,e2,0), (2,e3,0), (2,e2,0), (2,e1,0), (2,e3,0))
     * 事件去重分析
     * Vector((1,e3,100), (1,e2,0))
     * Vector((1,e2,100), (1,e1,0), (1,e3,0))
     * Vector((3,e2,100))
     * Vector((3,e3,100))
     * Vector((2,e3,100), (2,e2,0), (2,e1,0))
     *
     */


    println("================末次触点归因=================")
    rdd2.flatMap(tp=>{
      val guid = tp._1
      tp._2.map(list=>{
        // 事件去重  待归因事件只能做为一个位置的计算标准
        val list2 = list.distinct
        for (i <- 0 until list2.size) yield {
          if (i==list2.size-1){
            (guid , list2(i) , 100)  // 首次触点
          }else {
            (guid , list2(i) , 0)
          }
        }
      })
    }).foreach(println)

    println("================线性归因=================")
    rdd2.flatMap(tp=>{
      val guid = tp._1
      tp._2.map(list=>{
        // 事件去重  待归因事件只能做为一个位置的计算标准
      //  val list2 = list.distinct
        for (i <- 0 until list.size) yield {
          // 每个节点的权重一致  1/2     1/3  1/4
          (guid , list(i) , (1D/list.size)*100)
        }
      })
    }).foreach(println)

    println("================时间归因=================")

    /**
     * 数据随着时间的后移  前面的权重表小
     *    e3    e2    e3
     *    1      2    3     6
     *    1/6   2/6  3/6
     */

    rdd2.flatMap(tp=>{
      val guid = tp._1
      tp._2.map(list=>{
        // 事件去重  待归因事件只能做为一个位置的计算标准
       val fm =  (for (i<- 1 to list.size) yield i).sum
        //  val list2 = list.distinct
        for (i <- 0 until list.size) yield {  // 0  1  2  123
          // 每个节点的权重一致  1/2     1/3  1/4
          (guid , list(i) , ((i+1D)/fm )*100)
        }
      })
    }).foreach(println)

    println("================位置归因=================")

    /*
    * e1   e2  e3  e4  e5
    *  30  15  10   15  30
    *  e1   e2  e3  e4
    *  40  10  10  40
    *
    * 如果归因事件去重   归因列表最大 到 归因事件个数      4   5
    *   3
    * 100
    * 50 50
    * 40 20 40
    * 40  10  10  40
    * 30  15  10   15  30
    * e1  e2  e3  e5  e7  e6
    *
    * */

    rdd2.flatMap(tp=>{
      val guid = tp._1
      tp._2.map(list=>{
        // 确定一个数字列表   1 2 3 4 5 6
        val res1 =  (1 to list.size).toList
        // 确定一个数字列表   6 5 4 3 2 1
        val res2 = res1.reverse
        // 拉链操作 (1,6) , (2,5) , (3,4)  ,(4,3) (5,2) (6,1)
        val zip = res1.zip(res2).toList
        // 求绝对值
        val v = zip.map(tp => {
          // (5 3 1 1 3 5)
          Math.abs(tp._1 - tp._2)
        })
        // 求和 分母 计算占比
        val fm = v.sum
        // 事件列表和  权重列表 拉链
        val tuples = list.zip(v)
        tuples.map(tp=>{
          // 计算位置归因值
          (guid , tp._1 , (tp._2.toDouble/fm)*100)
        })
      })
    }).foreach(println)
    //  业务归因名   目标事件   归因类型 ,  e4  , guid , event_id, ratio
    //  创建报表 写入


  }
}
